package com.hl.wechat.demo.websocket;

import com.alibaba.fastjson.JSON;
import com.hl.wechat.demo.entity.SystemUser;
import com.hl.wechat.demo.entity.User;
import com.hl.wechat.demo.enums.MessageActionEnum;
import com.hl.wechat.demo.message.WebSocketMessageEntity;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import javax.websocket.Session;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Enumeration;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Consumer;
import java.util.stream.Collector;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import static com.hl.wechat.demo.websocket.WebSocketListener.webSocketManager;


public  class WebSocketManager implements IWebSocket {
    private Logger logger = LoggerFactory.getLogger(WebSocketManager.class);

    public WebSocketManager() {

    }

    /**
     * 保存WebSocket,key 是openId，value是Session
     */
    private ConcurrentHashMap<String, Session> concurrentMap = new ConcurrentHashMap<>();

    /**
     * 离线消息
     */
    private ConcurrentHashMap<String,List<WebSocketMessageEntity>> offlineMessage =new ConcurrentHashMap<>();

    @Autowired
    MessageSenderImpl messageSender;

    @Override
    public void addUser(String id, Session session) {
        logger.info("客户端连接{}",id);
        Session sessionIfAbsent = concurrentMap.putIfAbsent(id, session);
        if (sessionIfAbsent!=null){
            try {
                session.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        /**
         * 取出离线消息，并发送、删除
         */
        if (offlineMessage.get(id)!=null){
            List<WebSocketMessageEntity> webSocketMessageEntities = offlineMessage.get(id);
            if(messageSender.sendMessage(session, webSocketMessageEntities)){
                offlineMessage.remove(id);
            }
        }
        broadcastMessage(new SystemUser(),"flush-user");
    }

    @Override
    public void remove(String id) {
        concurrentMap.remove(id);
        broadcastMessage(new SystemUser(),"flush-user");
        logger.info("客户端断开{}",id);
    }

    @Override
    public boolean isOnline(String openId) {
        return concurrentMap.containsKey(openId);
    }

    @Override
    public void sendMessage(WebSocketMessageEntity entity) {
        Session session = concurrentMap.get(entity.getToId());
        if (null == session) {
            List<WebSocketMessageEntity> webSocketMessageEntities = offlineMessage.get(entity.getToId());
            if (webSocketMessageEntities == null) {
                webSocketMessageEntities = new ArrayList<>();
            }
            webSocketMessageEntities.add(entity);
            offlineMessage.put(entity.getToId(), webSocketMessageEntities);
            logger.info("{}未链接", entity);
            return;
        }
        messageSender.sendMessage(session, Stream.of(entity).collect(Collectors.toList()));
    }
    @Override
    public void broadcastMessage(User user, String msg) {
        /**
         * 构建WebSocketMessageEntity，并广播
         */
        messageSender.broadcastMessage(listSocketSession(),Stream.of(new WebSocketMessageEntity(user.getId(),
                "ALL_USER",msg, MessageActionEnum.BROADCAST.getValue())).collect(Collectors.toList()));
    }

    @Override
    public List<Session> listSocketSession() {
        Collection<Session> values = concurrentMap.values();
        return  new ArrayList<>(values);
    }

    @Override
    public List<String> listSessionId() {
        Enumeration<String> keys = concurrentMap.keys();
        List<String> result = new ArrayList<>();
        while (keys.hasMoreElements()) {
            result.add(keys.nextElement() + "");
        }
        return result;
    }

    @Override
    public User buildSystemUser() {
        return new SystemUser();
    }
}
